package cn.xiaolang.thread.supplier2consume;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

@Slf4j
public class ConsumerLambda<T> extends Thread {

    private final ExecutorService pool;
    private final BlockingQueue<T> blockingQueue;
    private final Consumer<T> consumer;

    @Setter
    private volatile boolean shouldRun = true;

    public ConsumerLambda() {
        this.pool = Executors.newFixedThreadPool(5);
        this.blockingQueue = new LinkedBlockingQueue<>();
        this.consumer = t -> log.info("消费：{}" , t);
    }

    public boolean addToQueue(T t) {
        return this.blockingQueue.add(t);
    }


    @Override
    public void run() {
        while (shouldRun) {
            pool.submit(() -> {
                try {
                    // 消费
                    consumer.accept(blockingQueue.take());
                } catch (InterruptedException e) {
                    /* Clean up whatever needs to be handled before interrupting  */
                    Thread.currentThread().interrupt();
                }
            });
        }
    }

}